package com.heima.article.stream.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.stream.processor.IHotArticleProcessor;
import com.heima.behavior.dto.ArticleStreamMessage;
import com.heima.behavior.dto.UpdateArticleMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

/**
 * @author: itheima
 * @create: 2022-04-06 15:00
 */
@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {


    /**
     * 实时流式计算逻辑：统计最近一段时间（10秒以内）每篇文章行为数据（被阅读，被点赞，被评论，被收藏）。将实时计算结果发送到目标话题中
     *
     * @param input 数据流中，文章操作行为JSON数据
     *              {"add":1,"articleId":1507605153761480706,"type":1} -->点赞文章
     *              {"add":1,"articleId":1507605153761480706,"type":2} -->评论文章
     *              监听到原始数据流中行为数据格式
     * @return 将实时计算结果发送到目标话题中。数据格式：{"articleId":1507605153761480706,view:10,like:9,comment:5,collect:5}
     */
    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {
        //1.处理原始数据流中数据。重新指定数据KEY（文章ID） 数据Val（文章操作行为数据）
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 重新改写 数据流中键值
             * @param key 数据中键
             * @param value 数据中值
             * @return 键值对对象
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //1.1 获取原始数据中JSON字符串值，将JSON转为java对象
                UpdateArticleMessage message = JSON.parseObject(value, UpdateArticleMessage.class);
                //1.2 重新设置数据Key以及Val
                return new KeyValue<String, String>(message.getArticleId().toString(), value);
            }
        });

        //2.对处理后数据进行分组，根据文章ID进行分组
        KGroupedStream<String, String> groupedStream = map.groupByKey();

        //3.设置时间窗口：统计最近60秒内数据流中每篇文章行为操作
        TimeWindowedKStream<String, String> windowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(60)));

        //4.设置聚合 统计10秒内当前这篇文章多次操作行为进行聚合 从而得到当前文章所有操作次数
        //4.1 设置聚合初始化对象
        Initializer<String> initializer = new Initializer() {
            @Override
            public String apply() {
                //第一条操作行为消息进入后，聚合结果为空
                return null;
            }
        };
        //4.2 设置对10秒某篇文章多次操作行为进行聚合
        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {
            /**
             * 每篇文章每个操作行为都会触发聚合。聚合后回调函数
             * @param key 数据流中数据Key 文章ID
             * @param value 数据流中数据val  用户操作行为   {"add":1,"articleId":1507605153761480706,"type":1}
             * @param aggregate 上次聚合结果  {"articleId":1507605153761480706,view:10,like:9,comment:5,collect:5}
             * @return 某篇文章聚合后结果10秒内历史数据汇总
             */
            @Override
            public String apply(String key, String value, String aggregate) {
                //4.2.1 声明聚合文章行为结果对象
                ArticleStreamMessage result = null;
                //4.2.2 判断上次聚合结果是否有值 有值：将行为增量值累加  没值：设置初始化值为0
                if (StringUtils.isBlank(aggregate)) {
                    result = new ArticleStreamMessage();
                    result.setArticleId(Long.parseLong(key));
                    result.setView(0);
                    result.setCollect(0);
                    result.setComment(0);
                    result.setLike(0);
                } else {
                    //说明上次聚合结果有值
                    result = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }
                //4.2.3 根据当前这次行为操作（点赞，阅读，评论，收藏）更新聚合文章结果对象
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Integer add = updateArticleMessage.getAdd();
                switch (updateArticleMessage.getType()) {
                    case 0:
                        //阅读
                        result.setView(result.getView() + add);
                        break;
                    case 1:
                        //点赞
                        result.setLike(result.getLike() + add);
                        break;
                    case 2:
                        //评论
                        result.setComment(result.getComment() + add);
                        break;
                    case 3:
                        //收藏
                        result.setCollect(result.getCollect() + add);
                        break;
                }
                return JSON.toJSONString(result);
            }
        };
        KTable<Windowed<String>, String> aggregate = windowedKStream.aggregate(initializer, aggregator);

        //5.将实时流式计算结果发送到执行目标话题中
        //5.1 将Windowed对象类型转为字符串类型
        KStream<String, String> kStream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue(key.key(), value);
            }
        });
        return kStream;
    }
}
